package com.xiaohu.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object WordCount {
  def main(args: Array[String]): Unit = {
    //创建spark配置文件对象
    val conf: SparkConf = new SparkConf()
    //设置运行模式
    //如果是本地local模式运行的话，需要设置setMaster
    //将来如果是集群进行，将这句话注释即可
    conf.setMaster("local")
    //设置spark作业的名字
    conf.setAppName("wordcount")

    //创建spark core上下文环境对象
    val sc: SparkContext = new SparkContext(conf)
    //===================================================================================

    //读取文件,每次读取一行
    //RDD是spark core中的核心数据结构，将来运行的时候，数据会在RDD之间流动，默认基于内存计算
    val linesRDD: RDD[String] = sc.textFile("spark/data/wcs/*")
    println(s"linesRDD的分区数：${linesRDD.getNumPartitions}")

    //一行数据根据分隔符分割
    val wordRDD: RDD[String] = linesRDD.flatMap(_.split("\\|"))
    println(s"wordRDD的分区数：${wordRDD.getNumPartitions}")


    //将每一个单词组成(word,1)
    val kvRDD: RDD[(String, Int)] = wordRDD.map((_, 1))
    println(s"kvRDD的分区数：${kvRDD.getNumPartitions}")

    val resRDD: RDD[(String, Int)] = kvRDD.reduceByKey(_ + _)

    //根据键进行分组
//    val kvRDD2: RDD[(String, Iterable[(String, Int)])] = kvRDD.groupBy(_._1)
//    println(s"kvRDD2的分区数：${kvRDD2.getNumPartitions}")
//
//    val resRDD: RDD[(String, Int)] = kvRDD2.map((e: (String, Iterable[(String, Int)])) => (e._1, e._2.size))
//    println(s"resRDD的分区数：${resRDD.getNumPartitions}")
    //打印
//    resRDD.foreach(println)

//    /**
//     * 链式调用
//     */
//    sc.textFile("spark/data/words.txt")
//      .flatMap(_.split("\\|"))
//      .map((_, 1))
//      .groupBy(_._1)
//      .map((e: (String, Iterable[(String, Int)])) => (e._1, e._2.size))
//      .foreach(println)

    //指定的是文件夹的路径
    //spark如果是local本地运行的话，会将本地文件系统看作一个hdfs文件系统
    resRDD.saveAsTextFile("spark/data/outdata2")

  }
}
